#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   DTC_test.py
@Time    :   2025/03/19 13:31:28
@Author  :   YangLiang 
@Version :   1.0
@Contact :   yangl@genvict.com
@License :   (C)Copyright 2022-2025
@Desc    :   None
'''

# here put the import lib
from cgitb import enable
from email.message import Message
from pytest_assume.plugin import assume
 
import pytest
import binascii
import logging
import time
import requests
from sympy import I
# from sympy import assuming
from commDriver.TS_CAN import *
import allure
import sys
from collections.abc import Iterable

from sqlalchemy import false
sys.dont_write_bytecode = True
import Set_up


@allure.severity('critical')
class Test_DTC_Phy(object):

    def dtc_time_check(self,Driver,ex_dtc_list,timeout=5):
        '''
        输入故障码和期望状态
        循环请求 19服务,当读到dtc码或超时5s仍然无DTC码时中断
        '''
        global total_time
        flag = True
        time_start = time.time()
        while flag:
            if time.time() - time_start <= timeout :
                uds_rep,size = Driver['UDS'].requst_response('19 02 7F',3,enable_print = False)
                act_dtc_list = []
                if len(uds_rep) >= 7:
                    for i in range(int((len(uds_rep)-3)/4)):
                        act_dtc_list = uds_rep[3+i:7+i]
                        i += 4
                        if set(ex_dtc_list).issubset(set(act_dtc_list)):
                            time_end = time.time()
                            total_time = round(time_end - time_start, 3)
                            readstr = binascii.b2a_hex(bytearray(uds_rep[3:]),' ')
                            logging.info('total time cost :{0} s ---,DTC_status<<<{1}>>>\n'.format(str(total_time),readstr))
                            flag = False
            else:
                flag = False
                time_end = time.time()
                total_time = round(time_end - time_start, 3)
                logging.warning('Overtime, No DTC code output!')
                with assume: assert total_time < timeout 
        return total_time


    
    def clearDTC(self, Driver, iStep,checkinitialized=True):
        # 清除DTC故障
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '14 FF FF FF', '54', 0, iStep, False, 'UDS', 'noClearFault')
        time.sleep(2)
        if checkinitialized:
            iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '19 02 7F', '59 02 FF', 0, iStep, False, 'UDS', 'noClearFault')
            iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        return iStep

    def searchDTC(self, Driver,req, DTC_Number,iStep, isTrue, MatureConditionsTime=1000, timeout=0.5):
        # 查询故障
        with allure.step(f"Step{iStep}: search DTC {binascii.b2a_hex(bytearray(DTC_Number), ' ')}"):
            time.sleep(MatureConditionsTime/1000)
            a,b,c=Driver['UDS'].search_DTC(req, DTC_Number, timeout)
            if isTrue:
                if (b + MatureConditionsTime/1000) > (MatureConditionsTime/1000 + 0.8):
                    logging.error(f"MatureConditionsTime = {MatureConditionsTime/1000 + 0.1}, realtime = {b + MatureConditionsTime/1000}")
                if a != True:
                    logging.error(f"No found DTC{binascii.b2a_hex(bytearray(DTC_Number), ' ')}")
                with assume: assert a == True and (b + MatureConditionsTime/1000) <= (MatureConditionsTime/1000 + 0.8)
            else:
                with assume: assert a == False
            iStep +=1
        return iStep

    '''设置信号值'''
    def setSignalValue(Self, Driver, message, signal, value, iStep):
        with allure.step(f'Step{iStep}: set the value of the {signal} in {message} to {value}'):
            Driver['CAN'].write_db_signal('Vehicle_CAN','BCM', message, signal, value)
            time.sleep(0.2)
            iStep += 1
            readValue = Driver['CAN'].read_db_signal('Vehicle_CAN','BCM', message, signal)
        return iStep
            

    def checkDTC(self, Driver, DTCNumber, message, cycleTime, missCount, iStep, signal='', faultValue=0, normalValue=0, settype='enable', DtcExist = True):
        req = [0x19, 0x02, 0x7F]

        DTC_Number_Exist = DTCNumber + [0x2F]
        DTC_Number_NoExist = DTCNumber + [0x2E]
        

        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        time.sleep(1.5)

        if settype == 'enable':
            # 停发报文
            iStep = Driver['CAN'].setSignalEanble(message, False, iStep)
        else:
            iStep = Driver['CAN'].setSignalValue(message, signal, faultValue, iStep)
        # 查询故障是否已经产生
        iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, DtcExist, cycleTime * missCount + 100)

        if settype == 'enable':
            # 恢复报文
            iStep = Driver['CAN'].setSignalEanble(message, True, iStep)
            if message == '0x90':
                time.sleep(2)
        else:
            iStep = Driver['CAN'].setSignalValue(message, signal, normalValue, iStep)
        # 查询故障是否已经恢复
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, DtcExist, cycleTime + 100, 1)
        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        iStep = self.searchDTC(Driver, req, DTCNumber, iStep, DtcExist, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTCNumber, iStep, False, 3000, 1)

        return iStep


    @allure.title('Test low-Circuit Voltage DTC in default session')
    @allure.description('Test low-Circuit Voltage DTC in default session with physical addressing')
    
    def test_LowCircuitVoltage_DTC_Default_Phy(self,action,Driver):
        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xF0,0x03,0x16]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        # 设置电压为8v 制造低压故障
        iStep = Driver['CAN'].SetVolt(Driver, 8, 1.5, iStep)
        # 查询低压故障是否已经产生
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)
        # 恢复电压为13v
        iStep = Driver['CAN'].SetVolt(Driver, 13, 1.5, iStep)
        # 查询故障是否已经恢复
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2E], iStep, True, 1000)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, True, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, False, 3000, 1)

    @allure.title('Test low Circuit Voltage boundary value generate DTC code in default session')
    @allure.description('Test low Circuit Voltage boundary DTC in default sessionin with physical addressing')
    
    def test_LowCircuitVoltage_boundary_DTC_Phy(self,action,Driver):

        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xF0,0x03,0x16]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)

        # 无故障到有故障
        for i in range(6):
            voltage = round(9.3 - i/10, 2)
            
            iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
            iStep = self.clearDTC(Driver, iStep)
            time.sleep(0.5)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 1.5, iStep)
            if voltage < 9:
                iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)
        
        # 有故障到无故障
        for i in range(6):
            voltage = round(8.7 + i/10, 2)
            iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
            iStep = self.clearDTC(Driver, iStep)
            time.sleep(1)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 1.5, iStep)
            if voltage < 9:
                iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)
        
        iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
        iStep = self.clearDTC(Driver, iStep)

    @allure.title('Test high-Circuit Voltage DTC in default session')
    @allure.description('Test low-Circuit Voltage DTC in default session with physical addressing')
    
    def test_HighCircuitVoltage_DTC_Default_Phy(self,action,Driver):
        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xF0,0x03,0x17]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        # 设置电压为8v 制造低压故障
        iStep = Driver['CAN'].SetVolt(Driver, 17, 1.5, iStep)
        # 查询低压故障是否已经产生
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)
        # 恢复电压为13v
        iStep = Driver['CAN'].SetVolt(Driver, 13, 1, iStep)
        # 查询故障是否已经恢复
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2E], iStep, True, 1000)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, True, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, False, 3000, 1)

    @allure.title('Test high Circuit Voltage boundary value generate DTC code in default session')
    @allure.description('Test high Circuit Voltage boundary DTC in default sessionin with physical addressing')
    
    def test_HighCircuitVoltage_boundary_DTC_Phy(self,action,Driver):

        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xF0,0x03,0x17]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)

        # 无故障到有故障
        for i in range(6):
            voltage = round(15.7 + i/10, 2)
            
            iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
            iStep = self.clearDTC(Driver, iStep)
            time.sleep(0.5)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 1.5, iStep)
            if voltage > 16:
                iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)
        
        # 有故障到无故障
        for i in range(6):
            voltage = round(16.3 - i/10, 2)
            iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
            iStep = self.clearDTC(Driver, iStep)
            time.sleep(1)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 1.5, iStep)
            if voltage > 16:
                iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)
        
        iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
        iStep = self.clearDTC(Driver, iStep)

    
    @allure.title('Test Lost Communication with XCD message in default session')
    @allure.description('Test Lost Communication with XCD message in default session with physical addressing')
    
    def test_LostCommunication_Default_Phy(self, action, Driver):
        DTCNumber = [0xD0,0x97,0x87]
        
        iStep = 1
        iStep = self.checkDTC(Driver, DTCNumber, 0x107, 500, 1, iStep)

    @allure.title('Test DLost Communication with XCD message in default session')
    @allure.description('Test DLost Communication with XCD message in default session with physical addressing')
    def test_DLostCommunication_Default_Phy(self, action, Driver):
        DTCNumber = [0xD0,0xA2,0x87]
        
        iStep = 1
        iStep = self.checkDTC(Driver, DTCNumber, 0x90, 1000, 1, iStep)

    @allure.title('Test high-Circuit Voltage DTC in default session')
    @allure.description('Test low-Circuit Voltage DTC in default session with physical addressing')
    def test_BUSOFF_DTC_Default_Phy(self,action,Driver):
        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xC0,0x7B,0x88]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        # 联通继电器端口6，制造CAN/CANFD I Bus Off故障
        iStep = Driver['CAN'].SetJiDianQiStatus(Driver, iStep, 6, 1)
        time.sleep(1)
        # 断开继电器端口6，恢复CAN/CANFD I Bus Off故障
        iStep = Driver['CAN'].SetJiDianQiStatus(Driver, iStep, 6, 0)
        time.sleep(1)
        # 查询BUSSOFF故障是否已经产生并恢复
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2E], iStep, True, 1000)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, True, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, False, 3000, 1)

    
    @allure.title('Test VIDNotConfigured DTC in default session')
    @allure.description('Test VIDNotConfigured DTC in default session with physical addressing')
    def test_VIDNotConfigured_DTC_Default_Phy(self,action,Driver):
        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xD3,0x05,0x55]

        iStep = 1
        logging.info('wait 10s')
        time.sleep(10)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '10 01', '50 01', 0, iStep)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '19 02 7F', '59 02 FF', 0, iStep)
        # 清除DTC故障
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '14 FF FF FF', '54', 0, iStep, False, 'UDS', 'noClearFault')
        logging.info('wait 12s')
        time.sleep(12)
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '19 02 7F', '59 02 FF', 0, iStep, False, 'UDS', 'noClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        # 停发报文
        iStep = Driver['CAN'].setSignalEanble(0x3E6, False, iStep)
        logging.info('wait 20s')
        time.sleep(20)
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)

        # 恢复报文
        iStep = Driver['CAN'].setSignalEanble(0x3E6, True, iStep)
        logging.info('wait 12s')
        time.sleep(12)
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2E], iStep, True, 1000)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)
        logging.info('wait 10s')
        time.sleep(10)

        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, True, 3000, 1)
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '14 FF FF FF', '54', 0, iStep, False, 'UDS', 'noClearFault')
        logging.info('wait 12s')
        time.sleep(12)
        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, False, 3000, 1)

    @allure.title('Test VID Invalid DTC in default session')
    @allure.description('Test VID Invalid DTC in default session with physical addressing')
    def test_VIDInvalid_DTC_Default_Phy(self,action,Driver):
        req = [0x19, 0x02, 0x7F]
        DTC_Number = [0xD3,0x05,0x56]

        iStep = 1
        logging.info('wait 10s')
        time.sleep(10)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '10 01', '50 01', 0, iStep)
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '19 02 7F', '59 02 FF', 0, iStep)
        # 清除DTC故障
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '14 FF FF FF', '54', 0, iStep, False, 'UDS', 'noClearFault')
        logging.info('wait 12s')
        time.sleep(12)
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '19 02 7F', '59 02 FF', 0, iStep, False, 'UDS', 'noClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        # 修改报文发送的VID
        Driver['CAN'].update_signal_data(0x3E6, index=[0,1],value=[0x88,0x88])
        logging.info('wait 20s')
        time.sleep(12)
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2F], iStep, True, 1000)

        # 恢复报文发送的VID
        Driver['CAN'].update_signal_data(0x3E6, index=[0,1],value=[0x01,0x01])
        logging.info('wait 12s')
        time.sleep(12)
        iStep = self.searchDTC(Driver, req, DTC_Number + [0x2E], iStep, True, 1000)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)
        logging.info('wait 10s')
        time.sleep(10)

        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, True, 3000, 1)
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, '14 FF FF FF', '54', 0, iStep, False, 'UDS', 'noClearFault')
        logging.info('wait 12s')
        time.sleep(12)
        iStep = self.searchDTC(Driver, req, DTC_Number, iStep, False, 3000, 1)

    
